package com.xuliugen.rocketmq.springboot.starter;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.xuliugen.rocketmq.springboot.starter.consumer.DefaultRocketMQListenerContainer;
import com.xuliugen.rocketmq.springboot.starter.consumer.RocketMQListener;
import com.xuliugen.rocketmq.springboot.starter.consumer.annotation.RocketMQMessageListener;
import com.xuliugen.rocketmq.springboot.starter.provider.RocketMQTemplate;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.impl.MQClientAPIImpl;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.Order;
import org.springframework.core.env.StandardEnvironment;
import org.springframework.util.Assert;

import javax.annotation.Resource;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicInteger;

import static com.xuliugen.rocketmq.springboot.starter.consumer.common.constants.RocketMQListenerContainerPropertyConstants.*;

/**
 * @author xuliugen
 */
@Configuration
@EnableConfigurationProperties(RocketMQProperties.class)
@ConditionalOnClass(MQClientAPIImpl.class)
@Order
@Slf4j
public class RocketMQAutoConfiguration {

    /**
     * 默认的MQ生产者
     * @param rocketMQProperties RocketMQ属性配置
     * @return MQ生产者对象
     */
    @Bean
    @ConditionalOnClass(DefaultMQProducer.class)
    @ConditionalOnMissingBean(DefaultMQProducer.class)
    @ConditionalOnProperty(prefix = "spring.rocketmq", value = {"nameServer", "producer.group"})
    public DefaultMQProducer mqProducer(RocketMQProperties rocketMQProperties) {
        RocketMQProperties.Producer producerConfig = rocketMQProperties.getProducer();
        String groupName = producerConfig.getGroup();
        Assert.hasText(groupName, "[spring.rocketmq.producer.group] must not be null");

        DefaultMQProducer producer = new DefaultMQProducer(producerConfig.getGroup());
        producer.setNamesrvAddr(rocketMQProperties.getNameServer());
        producer.setSendMsgTimeout(producerConfig.getSendMsgTimeout());
        producer.setRetryTimesWhenSendFailed(producerConfig.getRetryTimesWhenSendFailed());
        producer.setRetryTimesWhenSendAsyncFailed(producerConfig.getRetryTimesWhenSendAsyncFailed());
        producer.setMaxMessageSize(producerConfig.getMaxMessageSize());
        producer.setCompressMsgBodyOverHowmuch(producerConfig.getCompressMsgBodyOverHowmuch());
        producer.setRetryAnotherBrokerWhenNotStoreOK(producerConfig.isRetryAnotherBrokerWhenNotStoreOk());

        return producer;
    }

    @Bean
    @ConditionalOnClass(ObjectMapper.class)
    @ConditionalOnMissingBean(name = "rocketMQMessageObjectMapper")
    public ObjectMapper rocketMQMessageObjectMapper() {
        return new ObjectMapper();
    }

    /**
     * 条件注入消费者模板类RocketMQTemplate
     * @param mqProducer   注入DefaultMQProducer对象
     * @param objectMapper 注入ObjectMapper对象，非必须
     * @return
     */
    @Bean
    @ConditionalOnBean(DefaultMQProducer.class)
    @ConditionalOnMissingBean(name = "rocketMQTemplate")
    public RocketMQTemplate rocketMQTemplate(
            DefaultMQProducer mqProducer,
            @Autowired(required = false)
            @Qualifier("rocketMQMessageObjectMapper") ObjectMapper objectMapper) {
        RocketMQTemplate rocketMQTemplate = new RocketMQTemplate();
        rocketMQTemplate.setProducer(mqProducer);
        if (Objects.nonNull(objectMapper)) {
            rocketMQTemplate.setObjectMapper(objectMapper);
        }
        return rocketMQTemplate;
    }

    /**
     * ApplicationContextAware:当一个类实现了这个接口之后，这个类就可以方便获得ApplicationContext中的所有bean。
     * 换句话说，就是这个类可以直接获取Spring配置文件中所有有引用到的bean对象。
     * <p>
     * </p>
     * 当Spring容器启动的时候获取所有含有RocketMQMessageListener注解的Bean，对其装填属性，然后注入到Spring容器中；
     * <pre class="code">
     * @Slf4j
     * @Service
     * @RocketMQMessageListener(topic = "test-topic-2", consumerGroup = "my-consumer_test-topic-2")
     * public class MyConsumer implements RocketMQListener<OrderPaidEvent>{
     *     public void onMessage(OrderPaidEvent orderPaidEvent) {
     *         log.info("received orderPaidEvent: {}", orderPaidEvent);
     *     }
     * }
     * </pre>
     * <p>
     * </p>
     * 大致思路：
     * 1、当项目启动的时候从ApplicationContext中获取所有含有RocketMQMessageListener注解的Bean；
     * 2、验证这个Bean是否是实现了RocketMQListener接口的类，是的话继续处理，不是的话就抛出异常；
     * 3、由于一个类对应一个RocketMQConsumer消费Bean，所以一个Bean应该需要注入一个DefaultRocketMQListenerContainer对象，（DefaultRocketMQListenerContainer就是包含一个Consumer的容器）
     * 4、通过BeanDefinitionBuilder构造出DefaultRocketMQListenerContainer对象；
     * 5、为DefaultRocketMQListenerContainer对象设置属性值；
     * 6、通过DefaultListableBeanFactory将DefaultRocketMQListenerContainer对象注入到Spring容器中；
     * 7、获取刚才注入到Spring容器中的DefaultRocketMQListenerContainer对象；
     * 8、通过DefaultRocketMQListenerContainer对象启动Consumer；
     */
    @Configuration
    @ConditionalOnClass(DefaultMQPushConsumer.class)
    @EnableConfigurationProperties(RocketMQProperties.class)
    @ConditionalOnProperty(prefix = "spring.rocketmq", value = "nameServer")
    @Order
    public static class ListenerContainerConfiguration implements ApplicationContextAware, InitializingBean {

        private ConfigurableApplicationContext applicationContext;

        private AtomicInteger counter = new AtomicInteger(0);

        @Resource
        private StandardEnvironment environment;

        @Resource
        private RocketMQProperties rocketMQProperties;

        private ObjectMapper objectMapper;

        public ListenerContainerConfiguration() {
        }

        @Autowired(required = false)
        public ListenerContainerConfiguration(
                @Qualifier("rocketMQMessageObjectMapper") ObjectMapper objectMapper) {
            this.objectMapper = objectMapper;
        }

        @Override
        public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
            this.applicationContext = (ConfigurableApplicationContext) applicationContext;
        }

        @Override
        public void afterPropertiesSet() {
            //获取所有含有RocketMQMessageListener注解的Bean，然后填充属性，并将其注入到Spring容器中
            Map<String, Object> beans = this.applicationContext.getBeansWithAnnotation(RocketMQMessageListener.class);

            if (Objects.nonNull(beans)) {
                beans.forEach(this::registerContainer);
            }
        }

        /**
         * 注册Bean
         * @param beanName
         * @param bean
         */
        private void registerContainer(String beanName, Object bean) {
            //Determine the target class of the given bean instance which might be an AOP proxy.
            //Returns the target class for an AOP proxy or the plain class otherwise.
            Class<?> clazz = AopUtils.getTargetClass(bean);

            //检测bean是否是RocketMQListener接口的实现类
            if (!RocketMQListener.class.isAssignableFrom(bean.getClass())) {
                throw new IllegalStateException(clazz + " is not instance of " + RocketMQListener.class.getName());
            }

            RocketMQListener rocketMQListener = (RocketMQListener) bean;
            //获取RocketMQMessageListener注解
            RocketMQMessageListener annotation = clazz.getAnnotation(RocketMQMessageListener.class);
            //构造出来一个DefaultRocketMQListenerContainer对象
            BeanDefinitionBuilder beanBuilder = BeanDefinitionBuilder.rootBeanDefinition(DefaultRocketMQListenerContainer.class);

            //设置NameServer和Topic
            beanBuilder.addPropertyValue(PROP_NAMESERVER, rocketMQProperties.getNameServer());
            //使用environment.resolvePlaceholders(annotation.topic())获取Environment中对象的属性配置
            beanBuilder.addPropertyValue(PROP_TOPIC, environment.resolvePlaceholders(annotation.topic()));

            //设置与消费者相关的信息
            beanBuilder.addPropertyValue(PROP_CONSUMER_GROUP, environment.resolvePlaceholders(annotation.consumerGroup()));
            beanBuilder.addPropertyValue(PROP_CONSUME_MODE, annotation.consumeMode());
            beanBuilder.addPropertyValue(PROP_CONSUME_THREAD_MAX, annotation.consumeThreadMax());
            beanBuilder.addPropertyValue(PROP_MESSAGE_MODEL, annotation.messageModel());
            beanBuilder.addPropertyValue(PROP_SELECTOR_EXPRESS, environment.resolvePlaceholders(annotation.selectorExpress()));
            beanBuilder.addPropertyValue(PROP_SELECTOR_TYPE, annotation.selectorType());
            beanBuilder.addPropertyValue(PROP_ROCKETMQ_LISTENER, rocketMQListener);
            if (Objects.nonNull(objectMapper)) {
                beanBuilder.addPropertyValue(PROP_OBJECT_MAPPER, objectMapper);
            }
            beanBuilder.setDestroyMethodName(METHOD_DESTROY);

            //构造ContainerName格式
            String containerBeanName = String.format("%s_%s", DefaultRocketMQListenerContainer.class.getName(), counter.incrementAndGet());

            //注册Bean到Spring容器中
            DefaultListableBeanFactory beanFactory = (DefaultListableBeanFactory) applicationContext.getBeanFactory();
            beanFactory.registerBeanDefinition(containerBeanName, beanBuilder.getBeanDefinition());

            //获取刚才注入到Spring容器中的Bean
            DefaultRocketMQListenerContainer container = beanFactory.getBean(containerBeanName, DefaultRocketMQListenerContainer.class);

            if (!container.isStarted()) {
                try {
                    //启动container，也就是启动RocketMQPushConsumer
                    container.start();
                } catch (Exception e) {
                    log.error("started container failed. {}", container, e);
                    throw new RuntimeException(e);
                }
            }
            log.info("register rocketMQ listener to container, listenerBeanName:{}, containerBeanName:{}", beanName, containerBeanName);
        }
    }

}
